package cn.amoqi.springbootrocketmq.v2.service.impl;

import cn.amoqi.springbootrocketmq.v2.config.MqProperties;
import cn.amoqi.springbootrocketmq.v2.enums.MqTypeEnum;
import cn.amoqi.springbootrocketmq.v2.service.MqClient;
import io.netty.util.internal.StringUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * @Description: rocketClient
 * @author: zhuiFeng
 * @date: 2021年04月17日 10:04
 */
@Component
public class RocketMqClient implements MqClient {

    private final DefaultMQProducer producer;

    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final MqProperties mqProperties;

    public RocketMqClient(DefaultMQProducer producer, MqProperties mqProperties) {
        this.producer = producer;
        this.mqProperties = mqProperties;
    }

    /**
     * 发送消息
     *
     * @param topic 主题
     * @param content 内容
     * @return 是否发送成功
     */
    public boolean send(String topic, String content) {
        return send(topic, null, null, content, 0);
    }


    /**
     * 发送消息
     *
     * @param topic 主题
     * @return 是否发送成功
     */
    @Override
    public boolean send(String topic, String tags, String content) {
        return send(topic, tags, null, content, 0);
    }

    @Override
    public String support() {
        return MqTypeEnum.RocketMq.name();
    }

    @Override
    public boolean send(String topic, String tags, String keys, String content) {
        return send(topic, tags, keys, content, 0);
    }

    @Override
    public boolean send(String topic, String content, int delayTimeLevel) {
        return send(topic, null, null, content, delayTimeLevel);
    }

    @Override
    public boolean send(String topic, String tags, String content, int delayTimeLevel) {
        return send(topic, tags, null, content, delayTimeLevel);
    }

    /**
     * 发送延时消息
     * String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
     * 延时级别1对应延时1秒后发送消息
     *
     * @param topic 主题
     * @param tags 标签
     * @param content 内容
     * @param delayTimeLevel 延迟级别
     * @return 是否成功
     */
    public boolean send(String topic, String tags, String keys, String content, int delayTimeLevel) {
        Message message = new Message(topic, content.getBytes(StandardCharsets.UTF_8));
        if (keys != null && keys.length() > 0) {
            message.setKeys(keys);
        }
        if (tags != null && tags.length() > 0) {
            message.setTags(tags);
        }
        if (delayTimeLevel > 0) {
            message.setDelayTimeLevel(delayTimeLevel);
        }
        return send(message);
    }

    /**
     * 发送消息
     *
     * @param message 消息体
     * @return 是否发送成功
     */
    @Override
    public boolean send(Message message) {
        try {
            //producer.createTopic("broker-a","wotazi33",1);
            SendResult sendResult = producer.send(message);
            // 同步发送消息，只要不抛异常就是成功
            if (sendResult != null) {
                logger.info("Send mq message success. :" + sendResult.toString());
            }
            return true;
        } catch (Exception e) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
            logger.error("Send mq message failed. Topic is:" + message.getTopic());
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 开发环境创建消费者实例
     *
     * @return 创建消费者实例
     */
    @Override
    public DefaultMQPushConsumer createConsumer(String groupId, String topic, String subExpression) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupId);
        consumer.setNamesrvAddr(mqProperties.getUrl());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        if(!StringUtil.isNullOrEmpty(subExpression)){
            consumer.subscribe(topic, subExpression);
        }
        return consumer;
    }


}
